package cn.ac.iie.di.ban.data.exchange.runner.slave.task.runTask;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.StatusEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskOrderRequest;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskOrderResponse;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.data.RocketMQArgvData;
import cn.ac.iie.di.ban.data.exchange.runner.slave.task.checkTask.CheckSQLAndPermissionTask;
import cn.iie.jsqlparser.helper.JSQLParserHelper;

import java.util.ArrayList;

import org.apache.commons.lang.exception.ExceptionUtils;


import java.util.List;

/**
 * Created by Black-Laptop on 12/9/2017.
 */
public class SubscriptionTask {

    private int taskId;
    private String resultPath;
    private String consumerNameserver;

    private String errorMsg;
    private SubscriptionTaskThread subscriptionTaskThread;
    private JSQLParserHelper jsqlParserHelper;

    private String uniq_key;
    private String nameServer;

    public SubscriptionTask(TaskOrderRequest taskOrderRequest, String nameServer) throws Exception {
        this.nameServer = nameServer;
        this.uniq_key = taskOrderRequest.getUniq_key();
        init(taskOrderRequest);
    }

    public TaskOrderResponse doSubscriptionTask() {
        TaskOrderResponse taskOrderResponse;
        Boolean subscriptionTaskRunSuccess = subscriptionTask();

        if (subscriptionTaskRunSuccess) {
            taskOrderResponse = setSubscriptionTaskRunSuccess();
        } else {
            taskOrderResponse = setSubscriptionTaskRunFail();
        }
        taskOrderResponse.setDetail("");

        return taskOrderResponse;
    }

    private void init(TaskOrderRequest taskOrderRequest) throws Exception {
        this.taskId = Integer.valueOf(taskOrderRequest.getTask_id());
        this.resultPath = taskOrderRequest.getResult_path();
        this.consumerNameserver = nameServer;
        this.errorMsg = "";
        CheckSQLAndPermissionTask checkSQLAndPermissionTask = new CheckSQLAndPermissionTask(taskOrderRequest);
        TaskOrderResponse taskOrderResponse = checkSQLAndPermissionTask.checkSQLAndPermission(taskOrderRequest.getUniq_key());


        if (taskOrderResponse.getStatus() != StatusEnum.CHECK_SUCCESS) {
            throw new Exception("checkPermission error:" + taskOrderResponse.getDetail());
        }

        String sql = taskOrderResponse.getDetail();

        try {
            this.jsqlParserHelper = new JSQLParserHelper(sql);
        } catch (Exception e) {
            throw new Exception("sql error:" + sql);
        }
        setSql();
    }

    private void setSql() throws Exception {
//        System.out.println("sql1:" + jsqlParserHelper.getSQLString());
        List<String> fromList = jsqlParserHelper.getSQLTables();
        List<String> fromListAfter = new ArrayList<String>();
        for (int i = 0; i < fromList.size(); i++) {
            String[] froms = fromList.get(i).split("\\.");
            fromListAfter.add(froms[froms.length - 1]);
        }
        jsqlParserHelper.setSQLTables(fromListAfter);
//        System.out.println("sql2:" + jsqlParserHelper.getSQLString());

    }

    private TaskOrderResponse setSubscriptionTaskRunSuccess() {


        TaskOrderResponse taskOrderResponse = new TaskOrderResponse(uniq_key, StatusEnum.SUBSCRIPTION_TASK_SUCCESS, "");
        return taskOrderResponse;
    }

    private TaskOrderResponse setSubscriptionTaskRunFail() {

        TaskOrderResponse taskOrderResponse = new TaskOrderResponse(uniq_key, StatusEnum.SUBSCRIPTION_TASK_ERROR, errorMsg);
        return taskOrderResponse;
    }

    private Boolean subscriptionTask() {
        try {
            RocketMQArgvData rocketMQArgvData = getSubscriptionTaskRocketMQArgv();
            this.subscriptionTaskThread = new SubscriptionTaskThread(this.jsqlParserHelper, rocketMQArgvData);
            this.subscriptionTaskThread.start();
        } catch (Exception e) {
            errorMsg = ExceptionUtils.getFullStackTrace(e);
            return false;
        }

        return true;
    }

    private RocketMQArgvData getSubscriptionTaskRocketMQArgv() throws Exception {
        List<String> receiveTopic = jsqlParserHelper.getSQLTables();
        Boolean receiveTopicIsOne = receiveTopic.size() == 1;
        if (!receiveTopicIsOne) {
            throw new Exception("topic length greater than one");
        }

        RocketMQArgvData rocketMQArgvData = new RocketMQArgvData();
        rocketMQArgvData.setReceiveConsumerNameserver(this.consumerNameserver);
        rocketMQArgvData.setReceiveConsumerGroup("iie_di_subscription_" + taskId + "_" + receiveTopic.get(0));
        rocketMQArgvData.setReceiveConsumerTopic(receiveTopic.get(0));
        rocketMQArgvData.setSendConsumerNameserver(this.consumerNameserver);
        rocketMQArgvData.setSendConsumerTopic(this.resultPath);

        return rocketMQArgvData;
    }

    public Boolean getSubscriptionTaskThreadStatus() {
        return subscriptionTaskThread.isAlive();
    }

    public void stopSubscriptionTaskThread() {
        this.subscriptionTaskThread.stopThread();
        while (true) {
            if (!subscriptionTaskThread.isAlive()) {
                break;
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
